package com.transport.analysis.traffic;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.lib.db.DBWritable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;

public class TrafficRecordWritable implements Writable, DBWritable {
    private String recordId; // Can be bus_id or subway_id (String to accommodate both)
    private int routeId;
    private String stopId;
    private Timestamp timestamp;
    private String direction;
    private String status;
    private int passengersOn;
    private int passengersOff;
    private String actualSubwayId;
    private String actualBusId;

    public TrafficRecordWritable() {
    }

    public TrafficRecordWritable(String recordId, int routeId, String stopId, Timestamp timestamp,
            String direction, String status, int passengersOn, int passengersOff, String actualSubwayId,
            String actualBusId) {
        this.recordId = recordId;
        this.routeId = routeId;
        this.stopId = stopId;
        this.timestamp = timestamp;
        this.direction = direction;
        this.status = status;
        this.passengersOn = passengersOn;
        this.passengersOff = passengersOff;
        this.actualSubwayId = actualSubwayId;
        this.actualBusId = actualBusId;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(recordId);
        out.writeInt(routeId);
        out.writeUTF(stopId);
        out.writeLong(timestamp.getTime()); // Write timestamp as long
        out.writeUTF(direction);
        out.writeUTF(status);
        out.writeInt(passengersOn);
        out.writeInt(passengersOff);
        if (actualSubwayId == null) {
            out.writeBoolean(false);
        } else {
            out.writeBoolean(true);
            out.writeUTF(actualSubwayId);
        }
        if (actualBusId == null) {
            out.writeBoolean(false);
        } else {
            out.writeBoolean(true);
            out.writeUTF(actualBusId);
        }
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        recordId = in.readUTF();
        routeId = in.readInt();
        stopId = in.readUTF();
        timestamp = new Timestamp(in.readLong()); // Read timestamp as long
        direction = in.readUTF();
        status = in.readUTF();
        passengersOn = in.readInt();
        passengersOff = in.readInt();
        if (in.readBoolean()) {
            actualSubwayId = in.readUTF();
        } else {
            actualSubwayId = null;
        }
        if (in.readBoolean()) {
            actualBusId = in.readUTF();
        } else {
            actualBusId = null;
        }
    }

    @Override
    public void write(PreparedStatement statement) throws SQLException {
        statement.setString(1, recordId);
        statement.setInt(2, routeId);
        statement.setString(3, stopId);
        statement.setTimestamp(4, timestamp);
        statement.setString(5, direction);
        statement.setString(6, status);
        statement.setInt(7, passengersOn);
        statement.setInt(8, passengersOff);
        statement.setString(9, actualSubwayId);
        statement.setString(10, actualBusId);
    }

    @Override
    public void readFields(ResultSet resultSet) throws SQLException {
        this.recordId = resultSet.getString("bus_id") != null ? resultSet.getString("bus_id")
                : resultSet.getString("subway_id");
        this.routeId = resultSet.getInt("route_id");
        this.stopId = resultSet.getString("stop_id");
        this.timestamp = resultSet.getTimestamp("timestamp");
        this.direction = resultSet.getString("direction");
        this.status = resultSet.getString("status");
        this.passengersOn = resultSet.getInt("passengers_on");
        this.passengersOff = resultSet.getInt("passengers_off");
        this.actualSubwayId = resultSet.getString("subway_id");
        this.actualBusId = resultSet.getString("bus_id");
    }

    // Getters for Mapper
    public Timestamp getTimestamp() {
        return timestamp;
    }

    public int getPassengersOn() {
        return passengersOn;
    }

    public int getPassengersOff() {
        return passengersOff;
    }

    public String getStopId() {
        return stopId;
    }

    public int getRouteId() {
        return routeId;
    }

    public String getDirection() {
        return direction;
    }

    public String getStatus() {
        return status;
    }

    public String getSubwayId() {
        return actualSubwayId;
    }

    public String getBusId() {
        return actualBusId;
    }

    @Override
    public String toString() {
        return "TrafficRecordWritable{" +
                "recordId='" + recordId + '\'' +
                ", routeId=" + routeId +
                ", stopId='" + stopId + '\'' +
                ", timestamp=" + timestamp +
                ", direction='" + direction + '\'' +
                ", status='" + status + '\'' +
                ", passengersOn=" + passengersOn +
                ", passengersOff=" + passengersOff +
                ", actualSubwayId='" + actualSubwayId + '\'' +
                ", actualBusId='" + actualBusId + '\'' +
                '}';
    }
}